package com.javaspeak.concurrency.synchronizedperformance.contended;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Logger;
import org.junit.Test;
import org.junit.internal.TextListener;
import org.junit.runner.JUnitCore;

import com.javaspeak.concurrency.synchronizedperformance.contended.ContendedCounter;

/**
 * This test is about knowing the performance difference between using CAS and
 * and a synchronized method for updating a counter.
 * <p>
 * This test:
 * <ul>
 *    <li>tests updating a counter under contention using a synchronized method
 *    <li>test updating a counter under contention using a CAS AtomicInteger
 *    <li>validates that the counter was updated atomically
 *    <li>measures performance of the 2 methods of updating the counter and 
 *        shows how many times better the winner is.
 * </ul>
 * The results for jdk 1.7 on on a core i5 with 4 cores:
 * <ul>
 *    <li>Took 444 milli seconds to increment counter 4000000 times using 
 *        synchronized block and 4 concurrent calling threads
 *    <li>Took 116 milli seconds to increment counter 4000000 times using 
 *        CAS and 4 concurrent calling threads
 *    <li>Took 142 milli seconds to increment counter 4000000 times 
 *        using Disruptor's Sequence and 4 concurrent calling threads 
 * </ul>
 * The results were quite surprising. I was expecting the synchronized method
 * to perform far worse than the AtomicInteger which used CAS or the 
 * Disruptor Sequence which also used CAS. However AtomicInteger was only 3-4 
 * times faster than the synchronized method.
 * <p>
 * There was not a lot of difference between using the Disruptor's Sequence and
 * AtomicInteger on jdk 1.7 on on a core i5 with 4 cores:
 * <p>
 * The validation for updating the counter atomically checked the following:
 * <ul>
 *    <li>none of the 4 threads got the same value for incrementing the counter
 *    <li>no counter values were skipped.
 * <./ul>
 * This was achieved by getting each thread to remember the values it got back
 * from incrementing the counter in an array. When all threads had finished the
 * arrays of each thread were analyzed for duplicates.  A check was made for 
 * gaps by putting all the values in a sorted set and iterating through them 
 * and seeing that there were no gaps in the value.
 * 
 * @author John Dickerson
 */
public class ContendedSynchronizedTest {
    
    public Logger logger = Logger.getLogger( ContendedSynchronizedTest.class );
    

    private void verifyFutures( List<Future<int[]>> futures ) 
            throws ExecutionException, InterruptedException {
        
        Set<Integer> incrementSet = new TreeSet<Integer>();
        int[] increments;
        
        for ( Future<int[]> future : futures ){

            increments = future.get();

            for ( int j=0; j<increments.length; j++ ){

                if ( increments[ j ] != -1 ){

                    if ( incrementSet.contains( increments[ j ] ) ){

                        org.junit.Assert.fail( "duplicate" );
                    }
                    else {

                        incrementSet.add( increments[ j ] );
                    }
                }
            }
        }

        int numberElements = incrementSet.size();
        Iterator<Integer> numberIterator = incrementSet.iterator();

        // check for gaps
        for ( int i=0; i<numberElements; i++ ){

            org.junit.Assert.assertEquals(
                    new Integer( i ), numberIterator.next() );
        }
    }
     
    
    private long getTimeTaken( TestType testType, 
            int numberCallingThreads, 
                int numberCounterIncrementsPerThread, 
                        int sizeArrayReturned ) 
                             throws ExecutionException, InterruptedException {
                                 
         ContendedCounter contendedCounter = new ContendedCounter();
         
         ExecutorService executorService = 
                 Executors.newFixedThreadPool( numberCallingThreads );
         
         Collection<Callable<int[]>> callables = 
                 new ArrayList<Callable<int[]>>( numberCallingThreads );  
         
         for ( int i=0; i<numberCallingThreads; i++ ){
             
             Callable<int[]> callable;
             
             switch( testType ){
             
                 case CAS:
                     
                     callable = new CallableCallingCASMethod( 
                             numberCounterIncrementsPerThread, 
                                 contendedCounter, sizeArrayReturned );
                     break;
                     
                 case SYNCHRONIZED_METHOD: 
                     
                     callable = new CallableCallingSynchronizedMethod( 
                             numberCounterIncrementsPerThread, 
                                 contendedCounter, sizeArrayReturned );
                     break;
                     
                 case DISRUPTOR_SEQUENCE:
                     
                     callable = new CallableCallingDisruptorSequenceMethod( 
                             numberCounterIncrementsPerThread, 
                             contendedCounter, sizeArrayReturned );
                     break;
                     
                 default:
                     throw new RuntimeException( 
                             "Unexpected TestType: " + testType.name() );
             }
             
             callables.add( callable );
         }
         
         long startNano = System.nanoTime();
         
         List<Future<int[]>> futures = executorService.invokeAll( callables );
         
         long endNano = System.nanoTime();
         
         executorService.shutdown();
         
         verifyFutures( futures );
         
         return endNano - startNano;
    }
    
    
    @Test
    public void testContendedSynchronized() 
            throws ExecutionException, InterruptedException {
    
        int numberCallingThreads = 4;
        int numberCounterIncrementsPerThread = 1000 * 1000;
        
        int sizeArrayReturned = 
                numberCallingThreads * numberCounterIncrementsPerThread;
        
        long timeTakenSynchronizedMethodMilliseconds = 
                getTimeTaken( TestType.SYNCHRONIZED_METHOD,
                        numberCallingThreads, numberCounterIncrementsPerThread, 
                            sizeArrayReturned ) / 1000000;
        
        logger.debug( 
            "\n    Took " + timeTakenSynchronizedMethodMilliseconds + 
               " milli seconds to increment counter " + sizeArrayReturned + 
                   " times using synchronized block and " + 
                       numberCallingThreads +
                           " concurrent calling threads" );
        
        long timeTakenCasMethodMilliseconds =
                getTimeTaken( 
                    TestType.CAS,
                        numberCallingThreads, numberCounterIncrementsPerThread, 
                            sizeArrayReturned ) / 1000000;
        
        logger.debug( 
             "\n    Took " + timeTakenCasMethodMilliseconds + 
                " milli seconds to increment counter " + sizeArrayReturned + 
                    " times using CAS and " + 
                        numberCallingThreads +
                            " concurrent calling threads" );
        
        long timeTakenDisruptortSequenceMilliseconds =
                getTimeTaken( 
                     TestType.DISRUPTOR_SEQUENCE,   
                        numberCallingThreads, numberCounterIncrementsPerThread, 
                            sizeArrayReturned ) / 1000000;
        
        logger.debug( 
             "\n    Took " + timeTakenDisruptortSequenceMilliseconds + 
                " milli seconds to increment counter " + sizeArrayReturned + 
                    " times using Disruptor's Sequence and " + 
                        numberCallingThreads +
                            " concurrent calling threads" );
        
        if ( timeTakenSynchronizedMethodMilliseconds < timeTakenCasMethodMilliseconds && 
                timeTakenSynchronizedMethodMilliseconds < timeTakenDisruptortSequenceMilliseconds ){
            
            logger.debug( "Synchronized method wins for updating counter" );
        }
        else if ( timeTakenCasMethodMilliseconds < timeTakenSynchronizedMethodMilliseconds && 
                timeTakenCasMethodMilliseconds < timeTakenDisruptortSequenceMilliseconds ){
            
            logger.debug( "CAS method wins for updating counter" );
        }
        else if ( timeTakenDisruptortSequenceMilliseconds < timeTakenCasMethodMilliseconds &&
                timeTakenDisruptortSequenceMilliseconds < timeTakenSynchronizedMethodMilliseconds ){
            
            logger.debug( "Disruptor Sequence method wins for updating counter" );
        }
    }
    
    public static void main( String args[] ){
        
        BasicConfigurator.resetConfiguration();
        BasicConfigurator.configure();

        JUnitCore junitCore = new JUnitCore();
        junitCore.addListener( new TextListener( System.out ) );
        junitCore.run( ContendedSynchronizedTest.class );
    }
}
